package com.hliushi.spark.exmaple

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/30 10:26
 */
class SparkRDDPractice {

  //  12 张三 25 男 chinese 50
  //  12 张三 25 男 math 60
  //  12 张三 25 男 english 70
  //  12 李四 20 男 chinese 50
  //  12 李四 20 男 math 50
  //  12 李四 20 男 english 50
  //  12 王芳 19 女 chinese 70
  //  12 王芳 19 女 math 70
  //  12 王芳 19 女 english 70
  //  13 张大三 25 男 chinese 60
  //  13 张大三 25 男 math 60
  //  13 张大三 25 男 english 70
  //  13 李大四 20 男 chinese 50
  //  13 李大四 20 男 math 60
  //  13 李大四 20 男 english 50
  //  13 王小芳 19 女 chinese 70
  //  13 王小芳 19 女 math 80
  //  13 王小芳 19 女 english 70

  val spark: SparkSession = SparkSession.builder()
    .appName("spark_rdd_practice")
    .master("local[6]")
    .getOrCreate()
  val sc: SparkContext = spark.sparkContext

  sc.setLogLevel("ERROR")
  val file: RDD[String] = sc.textFile("dataset/interview_data/rdd_parctice")

  val rddStudent: RDD[Student] = file.map((line: String) => {
    val arr = line.split(" ")
    Student(arr(0).toInt, arr(1), arr(2).toInt, arr(3), arr(4), arr(5).toDouble)
  })

  // 1.1 一共有多少个小于20岁的人参加考试？
  @Test
  def test01(): Unit = {
    val result = rddStudent.filter((stu: Student) => stu.age < 20)
      .map((stu: Student) => stu.name)
      .distinct()
      .count()

    println(s"result = ${result}")
  }

  // 2. 一共有多个男生参加考试？
  @Test
  def test02(): Unit = {
    val result = rddStudent.filter((stu: Student) => stu.sex == "男")
      .map((stu: Student) => stu.name)
      .distinct()
      .count()

    println(s"result = ${result}")
  }

  //3. 12班有多少人参加考试？
  @Test
  def test03(): Unit = {
    val result = rddStudent.filter((stu: Student) => stu.classId == 12)
      .map((stu: Student) => stu.name)
      .distinct()
      .count()

    println(s"result = ${result}")
  }

  // 4. 语文科目的平均成绩是多少？
  @Test
  def test04(): Unit = {
    val count = rddStudent.map(stu => stu.name).distinct().count().toInt

    val result = rddStudent.filter(stu => stu.subject == "chinese")
      .map(stu => stu.score)
      .reduce((total: Double, curr: Double) => total + curr)


    val avg_score = result / count

    println(s"avg_score = ${avg_score}")
  }

  // 5.单个人平均成绩是多少？
  @Test
  def test05(): Unit = {

    val result = rddStudent
      .map((stu: Student) => stu.name -> (stu.score, 1))
      .reduceByKey((total: (Double, Int), curr: (Double, Int)) => (total._1 + curr._1, total._2 + curr._2))
      .map((stu: (String, (Double, Int))) => stu._1 -> stu._2._1 / stu._2._2)
    //.reduceByKey((total: Double, curr: Double) => total + curr)
    //.map((item: (String, Double)) => item._1 -> item._2 / count)
    result.collect().foreach(println)

  }

  // 6. 12班平均成绩是多少？
  @Test
  def test06(): Unit = {
    val result = rddStudent.filter(stu => stu.classId == 12)
      .map(stu => stu.classId -> (stu.score -> 1))
      .reduceByKey((total: (Double, Int), curr: (Double, Int)) => (total._1 + curr._1, total._2 + curr._2))
      .map((stu: (Int, (Double, Int))) => stu._1 -> stu._2._1 / stu._2._2)

    result.collect().foreach(println)

  }

  // 6. 12班男生平均总成绩是多少？
  @Test
  def test07(): Unit = {
    val result = rddStudent.filter(stu => stu.classId == 12 && stu.sex == "男")
      .map(stu => stu.classId -> (stu.score -> 1))
      .reduceByKey((total: (Double, Int), curr: (Double, Int)) => (total._1 + curr._1, total._2 + curr._2))
      .map((stu: (Int, (Double, Int))) => stu._1 -> stu._2._1 / stu._2._2)

    result.collect().foreach(println)

  }

  // 7. 12班语文成绩最低分是多少？
  @Test
  def test08(): Unit = {
    val minScore = rddStudent.filter(stu => stu.classId == 12 && stu.subject == "chinese")
      .map((stu: Student) => stu.score)
      .min()

    println(s"minScore = ${minScore}")

  }

  // 8. 总成绩大于150分的12班的女生有几个？
  @Test
  def test09(): Unit = {
    val result = rddStudent.filter(stu => stu.classId == 12 && stu.sex == "女")
      .map((stu: Student) => stu.name -> stu.score)
      .reduceByKey((total, curr) => total + curr)
      .filter((stu: (String, Double)) => stu._2 > 150)
      .count()


    println(s"result = ${result}")

  }
}

case class Student(classId: Int, name: String, age: Int, sex: String, subject: String, score: Double)